package zjs.dc.controller;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

/**
 * @Class: TAnnualWebsite
 * @Author: kongcb
 * @Description: 针对企查查t_annual_website表和中数t_an_websiteshop表进行数据映射开发
 * @Date: 2023/10/31 10:07
 */

public class TAnnualWebsite {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder().appName("t_annual_website").enableHiveSupport().getOrCreate();
        spark.sparkContext().setLogLevel("ERROR");
        Dataset<Row> zs_data = spark.sql("select * from a_dc_ep_ods.zs_t_an_websiteshop");
        long zs_count = zs_data.count();
        if(zs_count>0){
            spark.sql("select distinct * from (" +
                    "select id,company_id,company_name from a_dc_ep_dwi.t_annual_basicinfo " +
                    "union all " +
                    "select id,company_id,company_name from a_dc_ep_incr.t_annual_basicinfo) as a").createOrReplaceTempView("annual_basicinfo");
            //获取中数增量数据表中的新增数据（left join）
            spark.sql("insert overwrite table a_dc_ep_incr.t_annual_website select \n" +
                    "a.record_id as id,\n" +
                    "a.ancheid as annual_report_id,\n" +
                    "b.company_id as key_no,\n" +
                    "b.company_id,\n" +
                    "b.company_name,\n" +
                    "'' as `no`,\n" +
                    "a.webtype as type,\n" +
                    "a.websitname as name,\n" +
                    "a.domain as website,\n" +
                    "a.jobid as dates,\n" +
                    "if(b.id is null,'1','0') as isadd \n" +
                    "from a_dc_ep_ods.zs_t_an_websiteshop as a left join annual_basicinfo as b on \n" +
                    "a.ancheid=b.id");
            spark.sql("insert into table a_dc_ep_incr.t_annual_website select id,\n" +
                    "annual_report_id,\n" +
                    "key_no,\n" +
                    "company_id,\n" +
                    "company_name,\n" +
                    "`no`,\n" +
                    "type,\n" +
                    "name,\n" +
                    "website,\n" +
                    "dates,\n" +
                    "'-1' isadd from a_dc_ep_dwi.t_annual_website where annual_report_id in (select distinct ancheid from a_dc_ep_ods.zs_t_an_websiteshop_del)");
        }else {
            spark.sql("TRUNCATE TABLE a_dc_ep_incr.t_annual_website");
            System.out.println("中数本期数据涉及该表的数据为空。。。。。。");
        }

        spark.stop();
    }
}
